package com.springcloud.eshop.datasync.rabbitmq.strategy;

import com.alibaba.fastjson.JSON;
import com.springcloud.eshop.datasync.rabbitmq.RabbitmqProvider;
import com.springcloud.eshop.message.AggrDataChangeMessage;
import com.springcloud.eshop.message.DataChangeMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolExecutorFactoryBean;
import org.springframework.stereotype.Service;

import java.util.Set;
import java.util.concurrent.*;

/**
 * @Desc:
 * @Date: 2020-07-02
 * @Version:v1.0
 */
@Service
public abstract class BaseDataChangeHandler {

    private final static String AGGR_DATA_CHANGE_QUEUE = "aggr_data_change_queue";

    /**
     * 去重消息队列
     */
    private volatile Set<AggrDataChangeMessage> messageSet = new CopyOnWriteArraySet<>();

    /**
     * 批量商品Id
     */
    private volatile Set<Long> productIdSet = new CopyOnWriteArraySet<>();

    @Autowired
    private RabbitmqProvider rabbitmqProvider;

    private ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 5, 0, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(1000), Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy());

    public BaseDataChangeHandler() {
        new SendThread().start();
    }

    public abstract boolean handler(DataChangeMessage dataChangeMessage);

    public abstract String getRedisKey(long id);

    /**
     * @Desc: 发送数据聚合消息
     * @Param topic
     * @Param aggrDataChangeMessage
     * @Return void
     * @Date: 2020/7/2
     */
    public void sendMessage(AggrDataChangeMessage aggrDataChangeMessage) {
        rabbitmqProvider.sendMessage(AGGR_DATA_CHANGE_QUEUE, JSON.toJSONString(aggrDataChangeMessage));
    }

    /**
     * @Desc: 添加商品Id
     * @Param dataChangeMessage
     * @Return boolean
     * @Date: 2020/7/3
     */
    public boolean addProductId(long productId) {
        this.productIdSet.add(productId);
        return true;
    }

    /**
     * @Desc: 添加消息
     * @Param dataChangeMessage
     * @Return boolean
     * @Date: 2020/7/3
     */
    public boolean addMessage(AggrDataChangeMessage aggrDataChangeMessage) {
        this.messageSet.add(aggrDataChangeMessage);
        return true;
    }

    /**
     * @Desc: 获取商品Id集合
     * @Param
     * @Return java.util.Set<java.lang.Long>
     * @Date: 2020/7/4
     */
    public Set<Long> getProductIdSet() {
        return this.productIdSet;
    }

    /**
     * @Desc: 清空商品Id集合
     * @Param
     * @Return boolean
     * @Date: 2020/7/4
     */
    public boolean clearProductIdSet() {
        this.productIdSet.clear();
        return true;
    }

    private class SendThread extends Thread {
        @Override
        public void run() {
            while (true) {
                if (!messageSet.isEmpty()) {
                    for (AggrDataChangeMessage aggrDataChangeMessage : messageSet) {
                        sendMessage(aggrDataChangeMessage);
                        System.out.println("【将去重后的维度数据变更消息发送到下一个队列】， message:" + JSON.toJSONString(aggrDataChangeMessage));
                    }
                    messageSet.clear();
                }

                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
